package org.example.portrait.module;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.example.portrait.module.model.UserBehavior;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * 模拟用户行为数据生产者，向Kafka发送用户行为数据
 */
public class UserBehaviorProducer {
    private static final Logger logger = LoggerFactory.getLogger(UserBehaviorProducer.class);

    // Kafka主题名称，需与Flink程序中配置的一致
    private static final String TOPIC = "user-behavior-topic";
    // Kafka服务器地址
    private static final String BOOTSTRAP_SERVERS = "124.222.42.79:9092";

    // 模拟数据：用户ID范围
    private static final int USER_ID_RANGE = 3;
    // 模拟数据：商品ID范围
    private static final int ITEM_ID_RANGE = 100;
    // 模拟行为类型
    private static final String[] BEHAVIOR_TYPE = {"click", "purchase", "share"};
    // 模拟商品分类
    private static final String[] CATEGORIES = {"electronics", "clothing", "books"};

    public static void main(String[] args) {
        logger.info("用户行为数据生产者启动，目标主题: {}", TOPIC);

        // 配置Kafka生产者
        Properties props = new Properties();
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 增加生产环境常用配置
        props.put("acks", "all"); // 确保数据可靠性
        props.put("retries", 3);  // 失败重试次数
        props.put("linger.ms", 1); // 批处理延迟时间
        props.put("batch.size", 16384); // 批处理大小

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        Random random = new Random();

        try {
            // 持续发送模拟数据
            int count = 0;
            while (true) {
                // 生成随机用户行为数据
                UserBehavior behavior = generateRandomUserBehavior(random);

                // 转换为JSON字符串
                String jsonMessage = JSON.toJSONString(behavior);

                // 发送到Kafka
                ProducerRecord<String, String> record = new ProducerRecord<>(
                        TOPIC,
                        String.valueOf(behavior.getUserId()),  // 以用户ID为key，便于分区
                        jsonMessage
                );

                int finalCount = count;
                producer.send(record, (metadata, exception) -> {
                    if (exception == null) {
                        logger.info("发送成功 [{}]: 用户ID={}, 行为={}, 偏移量={}",
                                finalCount, behavior.getUserId(), behavior.getBehaviorType(), metadata.offset());
                    } else {
                        logger.error("发送失败 [{}]: {}", finalCount, exception.getMessage(), exception);
                    }
                });

                count++;
                // 每发送10条数据暂停一下，避免消息发送过快
                if (count % 10 == 0) {
                    logger.debug("已发送{}条数据，暂停30秒", count);
                    TimeUnit.SECONDS.sleep(30);
                }
            }
        } catch (InterruptedException e) {
            logger.warn("生产者被中断", e);
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            logger.error("发生未预期错误", e);
        } finally {
            logger.info("关闭Kafka生产者");
            producer.close();
        }
    }

    /**
     * 生成随机的用户行为数据
     */
    private static UserBehavior generateRandomUserBehavior(Random random) {
        UserBehavior behavior = new UserBehavior();
        behavior.setUserId(random.nextInt(USER_ID_RANGE) + 1);  // 1~USER_ID_RANGE的用户ID
        behavior.setItemId(random.nextInt(ITEM_ID_RANGE) + 1); // 1~ITEM_ID_RANGE的商品ID
        behavior.setBehaviorType(BEHAVIOR_TYPE[random.nextInt(BEHAVIOR_TYPE.length)]);
        behavior.setCategory(CATEGORIES[random.nextInt(CATEGORIES.length)]);
        behavior.setTimestamp(System.currentTimeMillis());  // 当前时间戳

        logger.trace("生成用户行为数据: {}", behavior);
        return behavior;
    }
}
